package com.firefightingiot.firefightingiot.config.rocketmq;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.firefightingiot.firefightingiot.config.scheduling.Scheduling;
import com.firefightingiot.firefightingiot.support.utils.MyLogger;
import com.firefightingiot.firefightingiot.web.service.DeviceService;
import com.firefightingiot.firefightingiot.web.service.FireService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

/**
 * @Author 18011618
 * @Date 10:31 2018/7/18
 * @Function 模拟用户消息消费
 */
@Component
public class UserConsumer {
    /**
     * 消费者的组名
     */
    @Value("${suning.rocketmq.conumerGroup}")
    private String consumerGroup;

    /**
     * NameServer 地址
     */
    @Value("${suning.rocketmq.namesrvaddr}")
    private String namesrvAddr;

    @Resource
    private Scheduling scheduling;
    @Resource
    public DeviceService deviceService;

    @Resource
    public MyLogger logger;
    @Resource
    FireService fireService;
    @PostConstruct
    public void consumer() {
        System.err.println("init defaultMQPushConsumer");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
        //consumer.setVipChannelEnabled(false);
        try {
            consumer.subscribe("user-topic", "user-tag");
            consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
                try {
                    for (MessageExt messageExt : list) {
                        String mes = new String(messageExt.getBody());
                        logger.info("consumer"+"------"+mes);
                        JSONObject jsonObject = JSONUtil.parseObj(mes);
                        String deviceFlag = jsonObject.getStr("deviceFlag");
                        if(deviceFlag.equals("updateToOff")){
                            scheduling.updateToOffMethod();
                        }else if(deviceFlag.equals("updateDeviceBusinessName")){
                            deviceService.updateBusinessName(jsonObject);
                        }else if(deviceFlag.equals("restRecoverFile")){
                            //fireService.restRecoverFile(jsonObject);
                        }else if(deviceFlag.equals("restFile")){
                            //fireService.restFile(jsonObject);
                        }else if(deviceFlag.equals("test")){
                            System.err.println("消费消息: " + "deviceFlag:"+deviceFlag+"-----"+mes);
                        }

                        //System.err.println("消费消息: " + new String(messageExt.getBody()));//输出消息内容
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再试
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
